
package cn.tang.tframe.threadpool;

import cn.tang.tframe.common.utils.thread.CustomeThreadPoolExecute;
import cn.tang.tframe.common.utils.thread.NamedThreadFactory;

import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author tanghc
 */
public class CommonThreadPoolManager {

	// 线程池维护线程的最少数量
	private static int CORE_SIZE = 200;

	// 线程池维护线程的最大数量
	private static int MAX_SIZE = 500;

	// 线程池维护线程所允许的空闲时间
	private static int KEEP_ALIVE_TIME = 0;

	// 线程池所使用的缓冲队列大小
	private static int WORK_QUEUE_SIZE = 100;

	// 任务调度周期
	private static int SCHEDULE_TASK_QOS_PERIOD = 10;
	// 任务调度周期
	private static int SCHEDULE_CORE_SIZE = 1;
	/*
	 * 创建一个调度线程池
	 */
	static ScheduledExecutorService scheduler = null;
	/*
	 * 线程池超出界线时将任务加入缓冲队列
	 */
	static final RejectedExecutionHandler handler =
		new RejectedExecutionHandler() {

			public void rejectedExecution(
				Runnable task, ThreadPoolExecutor executor) {

				taskQueue.offer(task);
			}
		};

	/*
	 * 将缓冲队列中的任务重新加载到线程池
	 */
	static final Runnable accessBufferThread = new Runnable() {

		public void run() {

			if (hasMoreAcquire()) {
				threadPool.execute(taskQueue.poll());
			}
		}
	};

	/*
	 * 线程池
	 */
	static ThreadPoolExecutor threadPool = null;


	// 任务缓冲队列
	private static Queue<Runnable> taskQueue = new LinkedList<>();
	static {
		threadPool = new CustomeThreadPoolExecute(
				CORE_SIZE, MAX_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
				new ArrayBlockingQueue<Runnable>(WORK_QUEUE_SIZE)
				,new NamedThreadFactory("QTFRAME")
				,handler);
		scheduler =
				Executors.newScheduledThreadPool(SCHEDULE_CORE_SIZE);
					/*
					 * 通过调度线程周期性的执行缓冲队列中任务
					 */
		scheduler.scheduleAtFixedRate(
				accessBufferThread, 0, SCHEDULE_TASK_QOS_PERIOD,
				TimeUnit.MILLISECONDS);
		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
			@Override
			public void run() {
				doShutDown();
			}
		}));
	}

	public void addThread(Runnable task) {

		addExecuteTask(task);
	}

	public void addThreads(List<? extends Runnable> tasks) {

		for (Runnable task : tasks) {
			addExecuteTask(task);
		}
	}

	/*
	 * 消息队列检查方法
	 */
	private static boolean hasMoreAcquire() {

		return !taskQueue.isEmpty();
	}

	/*
	 * 向线程池中添加任务方法
	 */
	public void addExecuteTask(Runnable task) {

		if (task != null) {
			threadPool.execute(task);
		}
	}

	public void shutdown() {

		doShutDown();
	}

	private static void doShutDown() {

		if (null != threadPool) {
			threadPool.shutdown();
			threadPool = null;
		}
		if (null != scheduler) {
			scheduler.shutdown();
			scheduler = null;
		}
	}


	/*
	 * 将构造方法访问修饰符设为私有，禁止任意实例化。
	 */
	private CommonThreadPoolManager() {

	}

	private static class QtThreadPoolManagerHolder {

		private static CommonThreadPoolManager instance = new CommonThreadPoolManager();
	}

	public static CommonThreadPoolManager getInstance() {

		return QtThreadPoolManagerHolder.instance;
	}
}
